package com.demo.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Properties;

@Slf4j
@Component
public class KafkaProducerClient {
    @Value("${spring.kafka.producer.bootstrap-servers}")
    private String servers;
    @Value("${spring.kafka.producer.max.request.size}")
    private int maxRequestSize;
    @Autowired
    private KafkaProducerCallBack<String,String> kafkaProducerCallBack;
    private Producer<String,String> producer;
    @PostConstruct
    public void initProducer(){
        Properties props = new Properties();
        props.put("bootstrap.servers", servers);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("max.request.size",maxRequestSize);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG,0);
        producer = new KafkaProducer<String, String>(props);
    }

    public void pushMsg(String topic, String key, String value) throws Exception {
        ProducerRecord<String, String> record = new ProducerRecord<>(topic,key,value);
        producer.send(record, (RecordMetadata metadata, Exception e) -> {
            if (e != null) {
                kafkaProducerCallBack.onError(record,e);
            }else {
                kafkaProducerCallBack.onSuccess(record);
            }
        });
    }

    @PreDestroy
    public void closeProducer(){
        producer.close();
    }

}